package com.hdu;

import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.Consumer;


public class SystemTimer implements Timer {

    private String executorName;


    private final ExecutorService TASK_EXECUTOR = Executors.newFixedThreadPool(
            1,
            (runnable) -> {
                Thread thread = new Thread(runnable);
                thread.setName("executor-" + executorName);
                thread.setDaemon(false);
                return thread;
            }
    );

    private final DelayQueue<TimerTaskList> DELAY_QUEUE = new DelayQueue<>();
    private final AtomicInteger TASK_COUNTER = new AtomicInteger(0);
    private final TimingWheel TIMING_WHEEL;

    private final ReentrantReadWriteLock READ_WRITE_LOCK = new ReentrantReadWriteLock();
    private final ReentrantReadWriteLock.ReadLock READ_LOCK = READ_WRITE_LOCK.readLock();
    private final ReentrantReadWriteLock.WriteLock WRITE_LOCK = READ_WRITE_LOCK.writeLock();

    private final Consumer<TimerTaskList.TimerTaskEntry> REINSERT = this::addTimerTaskEntry;

    public SystemTimer(String executorName, Long tickMs, Integer wheelSize, Long startMs) {
        this.executorName = executorName;
        this.TIMING_WHEEL = new TimingWheel(
                tickMs,
                wheelSize,
                startMs,
                TASK_COUNTER,
                DELAY_QUEUE
        );
    }

    @Override
    public void add(TimerTask timerTask) {
        READ_LOCK.lock();
        try {
            addTimerTaskEntry(new TimerTaskList.TimerTaskEntry(timerTask, timerTask.DELAYMS + System.currentTimeMillis()));
        } finally {
            READ_LOCK.unlock();
        }
    }

    private void addTimerTaskEntry(TimerTaskList.TimerTaskEntry timerTaskEntry) {
        if (!TIMING_WHEEL.addTask(timerTaskEntry)) {
            if (!timerTaskEntry.cancel()) {
                TASK_EXECUTOR.submit(timerTaskEntry.timerTask);
            }
        }
    }


    /**
     * 推动时钟
     *
     * @return
     */
    @Override
    public boolean advanceClock() {
        while (true) {
            TimerTaskList bucket;
            try {
                bucket = DELAY_QUEUE.take();
                WRITE_LOCK.lock();
                TIMING_WHEEL.advanceClock(bucket.getEXPIRATION());
                bucket.flush(REINSERT);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } finally {
                WRITE_LOCK.unlock();
            }
        }
    }

    @Override
    public int size() {
        return TASK_COUNTER.get();
    }

    @Override
    public void shutdown() {
        TASK_EXECUTOR.shutdown();
    }
}